// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <cstdint>
#include <sstream>
#include <string_view>
#include <type_traits>
#include <utility>
#include <vector>

#include <nebula/array/array_dict.h>
#include <nebula/array/builder_binary.h>
#include <nebula/array/builder_decimal.h>
#include <nebula/array/builder_dict.h>
#include <nebula/array/builder_nested.h>
#include <nebula/array/builder_primitive.h>
#include <nebula/array/builder_time.h>
#include <nebula/array/builder_union.h>
#include <nebula/core/chunked_array.h>
#include <nebula/ipc/json_simple.h>
#include <nebula/types/scalar.h>
#include <nebula/types/type_traits.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>
#include <turbo/numeric/float16.h>
#include <turbo/log/logging.h>
#include <nebula/util/value_parsing.h>

#include <nebula/json/rapidjson_defs.h>

#include <merak/json.h>


namespace nebula::ipc::json {

    using nebula::internal::ParseValue;


    namespace {

        constexpr auto kParseFlags = merak::json::kParseFullPrecisionFlag | merak::json::kParseNanAndInfFlag;

        const char *JsonTypeName(merak::json::Type json_type) {
            switch (json_type) {
                case merak::json::kNullType:
                    return "null";
                case merak::json::kFalseType:
                    return "false";
                case merak::json::kTrueType:
                    return "true";
                case merak::json::kObjectType:
                    return "object";
                case merak::json::kArrayType:
                    return "array";
                case merak::json::kStringType:
                    return "string";
                case merak::json::kNumberType:
                    return "number";
                default:
                    return "unknown";
            }
        }

        turbo::Status JSONTypeError(const char *expected_type, merak::json::Type json_type) {
            return turbo::invalid_argument_error("Expected ", expected_type, " or null, got JSON type ",
                                                 JsonTypeName(json_type));
        }

        class Converter {
        public:
            virtual ~Converter() = default;

            virtual turbo::Status init() { return turbo::OkStatus(); }

            virtual turbo::Status AppendValue(const merak::json::Value &json_obj) = 0;

            turbo::Status append_null() { return this->builder()->append_null(); }

            virtual turbo::Status append_values(const merak::json::Value &json_array) = 0;

            virtual std::shared_ptr<ArrayBuilder> builder() = 0;

            virtual turbo::Status finish(std::shared_ptr<Array> *out) {
                auto builder = this->builder();
                if (builder->length() == 0) {
                    // Make sure the builder was initialized
                    TURBO_RETURN_NOT_OK(builder->resize(1));
                }
                return builder->finish(out);
            }

        protected:
            std::shared_ptr<DataType> type_;
        };

        turbo::Status GetConverter(const std::shared_ptr<DataType> &, std::shared_ptr<Converter> *out);

        // CRTP
        template<class Derived>
        class ConcreteConverter : public Converter {
        public:
            turbo::Result<int64_t> SizeOfJSONArray(const merak::json::Value &json_obj) {
                if (!json_obj.is_array()) {
                    return JSONTypeError("array", json_obj.get_type());
                }
                return json_obj.size();
            }

            turbo::Status append_values(const merak::json::Value &json_array) final {
                auto self = static_cast<Derived *>(this);
                TURBO_MOVE_OR_RAISE(auto size, SizeOfJSONArray(json_array));
                for (uint32_t i = 0; i < size; ++i) {
                    TURBO_RETURN_NOT_OK(self->AppendValue(json_array[i]));
                }
                return turbo::OkStatus();
            }

            const std::shared_ptr<DataType> &get_value_type() {
                if (type_->id() != Type::DICTIONARY) {
                    return type_;
                }
                return turbo::checked_cast<const DictionaryType &>(*type_).get_value_type();
            }

            template<typename BuilderType>
            turbo::Status MakeConcreteBuilder(std::shared_ptr<BuilderType> *out) {
                std::unique_ptr<ArrayBuilder> builder;
                TURBO_RETURN_NOT_OK(MakeBuilder(default_memory_pool(), this->type_, &builder));
                *out = turbo::checked_pointer_cast<BuilderType>(std::move(builder));
                DKCHECK(*out);
                return turbo::OkStatus();
            }
        };

        // ------------------------------------------------------------------------
        // Converter for null arrays

        class NullConverter final : public ConcreteConverter<NullConverter> {
        public:
            explicit NullConverter(const std::shared_ptr<DataType> &type) {
                type_ = type;
                builder_ = std::make_shared<NullBuilder>();
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return append_null();
                }
                return JSONTypeError("null", json_obj.get_type());
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<NullBuilder> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for boolean arrays

        class BooleanConverter final : public ConcreteConverter<BooleanConverter> {
        public:
            explicit BooleanConverter(const std::shared_ptr<DataType> &type) {
                type_ = type;
                builder_ = std::make_shared<BooleanBuilder>();
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return append_null();
                }
                if (json_obj.is_bool()) {
                    return builder_->append(json_obj.get_bool());
                }
                if (json_obj.is_int32()) {
                    return builder_->append(json_obj.get_int32() != 0);
                }
                return JSONTypeError("boolean", json_obj.get_type());
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BooleanBuilder> builder_;
        };

        // ------------------------------------------------------------------------
        // Helpers for numeric converters

        // Convert single signed integer value (also {Date,Time}{32,64} and Timestamp)
        template<typename T>
        enable_if_physical_signed_integer<T, turbo::Status> ConvertNumber(const merak::json::Value &json_obj,
                                                                          const DataType &type,
                                                                          typename T::c_type *out) {
            if (json_obj.is_int64()) {
                int64_t v64 = json_obj.get_int64();
                *out = static_cast<typename T::c_type>(v64);
                if (*out == v64) {
                    return turbo::OkStatus();
                } else {
                    return turbo::invalid_argument_error("Value ", v64, " out of bounds for ", type);
                }
            } else {
                *out = static_cast<typename T::c_type>(0);
                return JSONTypeError("signed int", json_obj.get_type());
            }
        }

        // Convert single unsigned integer value
        template<typename T>
        enable_if_unsigned_integer<T, turbo::Status> ConvertNumber(const merak::json::Value &json_obj,
                                                                   const DataType &type,
                                                                   typename T::c_type *out) {
            if (json_obj.is_uint64()) {
                uint64_t v64 = json_obj.get_uint64();
                *out = static_cast<typename T::c_type>(v64);
                if (*out == v64) {
                    return turbo::OkStatus();
                } else {
                    return turbo::invalid_argument_error("Value ", v64, " out of bounds for ", type);
                }
            } else {
                *out = static_cast<typename T::c_type>(0);
                return JSONTypeError("unsigned int", json_obj.get_type());
            }
        }

        // Convert float16/Fp16Type
        template<typename T>
        enable_if_half_float<T, turbo::Status> ConvertNumber(const merak::json::Value &json_obj,
                                                             const DataType &type, uint16_t *out) {
            if (json_obj.is_double()) {
                double f64 = json_obj.get_double();
                *out = turbo::Float16(f64).bits();
                return turbo::OkStatus();
            } else if (json_obj.is_uint32()) {
                uint32_t u32t = json_obj.get_uint32();
                auto f64 = static_cast<double>(u32t);
                *out = turbo::Float16(f64).bits();
                return turbo::OkStatus();
            } else if (json_obj.is_int32()) {
                int32_t i32t = json_obj.get_int32();
                auto f64 = static_cast<double>(i32t);
                *out = turbo::Float16(f64).bits();
                return turbo::OkStatus();
            } else {
                *out = static_cast<uint16_t>(0);
                return JSONTypeError("unsigned int", json_obj.get_type());
            }
        }

        // Convert single floating point value
        template<typename T>
        enable_if_physical_floating_point<T, turbo::Status> ConvertNumber(const merak::json::Value &json_obj,
                                                                          const DataType &type,
                                                                          typename T::c_type *out) {
            if (json_obj.is_number()) {
                *out = static_cast<typename T::c_type>(json_obj.get_double());
                return turbo::OkStatus();
            } else {
                *out = static_cast<typename T::c_type>(0);
                return JSONTypeError("number", json_obj.get_type());
            }
        }

        // ------------------------------------------------------------------------
        // Converter for int arrays

        template<typename Type, typename BuilderType = typename TypeTraits<Type>::BuilderType>
        class IntegerConverter final
                : public ConcreteConverter<IntegerConverter<Type, BuilderType>> {
            using c_type = typename Type::c_type;

            static constexpr auto is_signed = std::is_signed<c_type>::value;

        public:
            explicit IntegerConverter(const std::shared_ptr<DataType> &type) { this->type_ = type; }

            turbo::Status init() override { return this->MakeConcreteBuilder(&builder_); }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                c_type value;
                TURBO_RETURN_NOT_OK(ConvertNumber<Type>(json_obj, *this->type_, &value));
                return builder_->append(value);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for float arrays

        template<typename Type, typename BuilderType = typename TypeTraits<Type>::BuilderType>
        class FloatConverter final : public ConcreteConverter<FloatConverter<Type, BuilderType>> {
            using c_type = typename Type::c_type;

        public:
            explicit FloatConverter(const std::shared_ptr<DataType> &type) { this->type_ = type; }

            turbo::Status init() override { return this->MakeConcreteBuilder(&builder_); }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                c_type value;
                TURBO_RETURN_NOT_OK(ConvertNumber<Type>(json_obj, *this->type_, &value));
                return builder_->append(value);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for decimal arrays

        template<typename DecimalSubtype, typename DecimalValue, typename BuilderType>
        class DecimalConverter final
                : public ConcreteConverter<
                        DecimalConverter<DecimalSubtype, DecimalValue, BuilderType>> {
        public:
            explicit DecimalConverter(const std::shared_ptr<DataType> &type) {
                this->type_ = type;
                decimal_type_ = &turbo::checked_cast<const DecimalSubtype &>(*this->get_value_type());
            }

            turbo::Status init() override { return this->MakeConcreteBuilder(&builder_); }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                if (json_obj.is_string()) {
                    int32_t precision, scale;
                    DecimalValue d;
                    auto view = std::string_view(json_obj.get_string(), json_obj.get_string_length());
                    TURBO_RETURN_NOT_OK(DecimalValue::from_string(view, &d, &precision, &scale));
                    if (scale != decimal_type_->scale()) {
                        return turbo::invalid_argument_error("Invalid scale for decimal: expected ",
                                                             decimal_type_->scale(), ", got ", scale);
                    }
                    return builder_->append(d);
                }
                return JSONTypeError("decimal string", json_obj.get_type());
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
            const DecimalSubtype *decimal_type_;
        };

        template<typename BuilderType = typename TypeTraits<Decimal128Type>::BuilderType>
        using Decimal128Converter = DecimalConverter<Decimal128Type, Decimal128, BuilderType>;
        template<typename BuilderType = typename TypeTraits<Decimal256Type>::BuilderType>
        using Decimal256Converter = DecimalConverter<Decimal256Type, Decimal256, BuilderType>;

        // ------------------------------------------------------------------------
        // Converter for timestamp arrays

        class TimestampConverter final : public ConcreteConverter<TimestampConverter> {
        public:
            explicit TimestampConverter(const std::shared_ptr<DataType> &type)
                    : timestamp_type_{turbo::checked_cast<const TimestampType *>(type.get())} {
                this->type_ = type;
                builder_ = std::make_shared<TimestampBuilder>(type, default_memory_pool());
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                int64_t value;
                if (json_obj.is_number()) {
                    TURBO_RETURN_NOT_OK(ConvertNumber<Int64Type>(json_obj, *this->type_, &value));
                } else if (json_obj.is_string()) {
                    std::string_view view(json_obj.get_string(), json_obj.get_string_length());
                    if (!ParseValue(*timestamp_type_, view.data(), view.size(), &value)) {
                        return turbo::invalid_argument_error("couldn't parse timestamp from ", view);
                    }
                } else {
                    return JSONTypeError("timestamp", json_obj.get_type());
                }
                return builder_->append(value);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            const TimestampType *timestamp_type_;
            std::shared_ptr<TimestampBuilder> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for day-time interval arrays

        class DayTimeIntervalConverter final
                : public ConcreteConverter<DayTimeIntervalConverter> {
        public:
            explicit DayTimeIntervalConverter(const std::shared_ptr<DataType> &type) {
                this->type_ = type;
                builder_ = std::make_shared<DayTimeIntervalBuilder>(default_memory_pool());
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                DayTimeIntervalType::DayMilliseconds value;
                if (!json_obj.is_array()) {
                    return JSONTypeError("array", json_obj.get_type());
                }
                if (json_obj.size() != 2) {
                    return turbo::invalid_argument_error(
                            "day time interval pair must have exactly two elements, had ", json_obj.size());
                }
                TURBO_RETURN_NOT_OK(ConvertNumber<Int32Type>(json_obj[0], *this->type_, &value.days));
                TURBO_RETURN_NOT_OK(
                        ConvertNumber<Int32Type>(json_obj[1], *this->type_, &value.milliseconds));
                return builder_->append(value);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<DayTimeIntervalBuilder> builder_;
        };

        class MonthDayNanoIntervalConverter final
                : public ConcreteConverter<MonthDayNanoIntervalConverter> {
        public:
            explicit MonthDayNanoIntervalConverter(const std::shared_ptr<DataType> &type) {
                this->type_ = type;
                builder_ = std::make_shared<MonthDayNanoIntervalBuilder>(default_memory_pool());
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                MonthDayNanoIntervalType::MonthDayNanos value;
                if (!json_obj.is_array()) {
                    return JSONTypeError("array", json_obj.get_type());
                }
                if (json_obj.size() != 3) {
                    return turbo::invalid_argument_error(
                            "month_day_nano_interval  must have exactly 3 elements, had ", json_obj.size());
                }
                TURBO_RETURN_NOT_OK(ConvertNumber<Int32Type>(json_obj[0], *this->type_, &value.months));
                TURBO_RETURN_NOT_OK(ConvertNumber<Int32Type>(json_obj[1], *this->type_, &value.days));
                TURBO_RETURN_NOT_OK(
                        ConvertNumber<Int64Type>(json_obj[2], *this->type_, &value.nanoseconds));

                return builder_->append(value);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<MonthDayNanoIntervalBuilder> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for binary and string arrays

        template<typename Type, typename BuilderType = typename TypeTraits<Type>::BuilderType>
        class StringConverter final
                : public ConcreteConverter<StringConverter<Type, BuilderType>> {
        public:
            explicit StringConverter(const std::shared_ptr<DataType> &type) { this->type_ = type; }

            turbo::Status init() override { return this->MakeConcreteBuilder(&builder_); }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                if (json_obj.is_string()) {
                    auto view = std::string_view(json_obj.get_string(), json_obj.get_string_length());
                    return builder_->append(view);
                } else {
                    return JSONTypeError("string", json_obj.get_type());
                }
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for fixed-size binary arrays

        template<typename BuilderType = typename TypeTraits<FixedSizeBinaryType>::BuilderType>
        class FixedSizeBinaryConverter final
                : public ConcreteConverter<FixedSizeBinaryConverter<BuilderType>> {
        public:
            explicit FixedSizeBinaryConverter(const std::shared_ptr<DataType> &type) {
                this->type_ = type;
            }

            turbo::Status init() override { return this->MakeConcreteBuilder(&builder_); }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                if (json_obj.is_string()) {
                    auto view = std::string_view(json_obj.get_string(), json_obj.get_string_length());
                    if (view.length() != static_cast<size_t>(builder_->byte_width())) {
                        std::stringstream ss;
                        ss << "Invalid string length " << view.length() << " in JSON input for "
                           << this->type_->to_string();
                        return turbo::invalid_argument_error(ss.str());
                    }
                    return builder_->append(view);
                } else {
                    return JSONTypeError("string", json_obj.get_type());
                }
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
        };

        // ------------------------------------------------------------------------
        // Converter for list arrays

        template<typename TYPE>
        class VarLengthListLikeConverter final
                : public ConcreteConverter<VarLengthListLikeConverter<TYPE>> {
        public:
            using BuilderType = typename TypeTraits<TYPE>::BuilderType;

            explicit VarLengthListLikeConverter(const std::shared_ptr<DataType> &type) {
                this->type_ = type;
            }

            turbo::Status init() override {
                const auto &var_length_list_like_type = turbo::checked_cast<const TYPE &>(*this->type_);
                TURBO_RETURN_NOT_OK(
                        GetConverter(var_length_list_like_type.get_value_type(), &child_converter_));
                auto child_builder = child_converter_->builder();
                builder_ =
                        std::make_shared<BuilderType>(default_memory_pool(), child_builder, this->type_);
                return turbo::OkStatus();
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                // Extend the child converter with this JSON array
                TURBO_MOVE_OR_RAISE(auto size, this->SizeOfJSONArray(json_obj));
                TURBO_RETURN_NOT_OK(builder_->append(true, size));
                return child_converter_->append_values(json_obj);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<BuilderType> builder_;
            std::shared_ptr<Converter> child_converter_;
        };

        // ------------------------------------------------------------------------
        // Converter for map arrays

        class MapConverter final : public ConcreteConverter<MapConverter> {
        public:
            explicit MapConverter(const std::shared_ptr<DataType> &type) { type_ = type; }

            turbo::Status init() override {
                const auto &map_type = turbo::checked_cast<const MapType &>(*type_);
                TURBO_RETURN_NOT_OK(GetConverter(map_type.key_type(), &key_converter_));
                TURBO_RETURN_NOT_OK(GetConverter(map_type.item_type(), &item_converter_));
                auto key_builder = key_converter_->builder();
                auto item_builder = item_converter_->builder();
                builder_ = std::make_shared<MapBuilder>(default_memory_pool(), key_builder,
                                                        item_builder, type_);
                return turbo::OkStatus();
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                TURBO_RETURN_NOT_OK(builder_->append());
                if (!json_obj.is_array()) {
                    return JSONTypeError("array", json_obj.get_type());
                }
                auto size = json_obj.size();
                for (uint32_t i = 0; i < size; ++i) {
                    const auto &json_pair = json_obj[i];
                    if (!json_pair.is_array()) {
                        return JSONTypeError("array", json_pair.get_type());
                    }
                    if (json_pair.size() != 2) {
                        return turbo::invalid_argument_error("key item pair must have exactly two elements, had ",
                                                             json_pair.size());
                    }
                    if (json_pair[0].is_null()) {
                        return turbo::invalid_argument_error("null key is invalid");
                    }
                    TURBO_RETURN_NOT_OK(key_converter_->AppendValue(json_pair[0]));
                    TURBO_RETURN_NOT_OK(item_converter_->AppendValue(json_pair[1]));
                }
                return turbo::OkStatus();
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<MapBuilder> builder_;
            std::shared_ptr<Converter> key_converter_, item_converter_;
        };

        // ------------------------------------------------------------------------
        // Converter for fixed size list arrays

        class FixedSizeListConverter final : public ConcreteConverter<FixedSizeListConverter> {
        public:
            explicit FixedSizeListConverter(const std::shared_ptr<DataType> &type) { type_ = type; }

            turbo::Status init() override {
                const auto &list_type = turbo::checked_cast<const FixedSizeListType &>(*type_);
                list_size_ = list_type.list_size();
                TURBO_RETURN_NOT_OK(GetConverter(list_type.get_value_type(), &child_converter_));
                auto child_builder = child_converter_->builder();
                builder_ = std::make_shared<FixedSizeListBuilder>(default_memory_pool(),
                                                                  child_builder, type_);
                return turbo::OkStatus();
            }

            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                TURBO_RETURN_NOT_OK(builder_->append());
                // Extend the child converter with this JSON array
                TURBO_RETURN_NOT_OK(child_converter_->append_values(json_obj));
                if (json_obj.get_array().size() != static_cast<merak::json::SizeType>(list_size_)) {
                    return turbo::invalid_argument_error("incorrect list size ", json_obj.get_array().size());
                }
                return turbo::OkStatus();
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            int32_t list_size_;
            std::shared_ptr<FixedSizeListBuilder> builder_;
            std::shared_ptr<Converter> child_converter_;
        };

        // ------------------------------------------------------------------------
        // Converter for struct arrays

        class StructConverter final : public ConcreteConverter<StructConverter> {
        public:
            explicit StructConverter(const std::shared_ptr<DataType> &type) { type_ = type; }

            turbo::Status init() override {
                std::vector<std::shared_ptr<ArrayBuilder>> child_builders;
                for (const auto &field: type_->fields()) {
                    std::shared_ptr<Converter> child_converter;
                    TURBO_RETURN_NOT_OK(GetConverter(field->type(), &child_converter));
                    child_converters_.push_back(child_converter);
                    child_builders.push_back(child_converter->builder());
                }
                builder_ = std::make_shared<StructBuilder>(type_, default_memory_pool(),
                                                           std::move(child_builders));
                return turbo::OkStatus();
            }

            // append a JSON value that is either an array of N elements in order
            // or an object mapping struct names to values (omitted struct members
            // are mapped to null).
            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                if (json_obj.is_array()) {
                    auto size = json_obj.size();
                    auto expected_size = static_cast<uint32_t>(type_->num_fields());
                    if (size != expected_size) {
                        return turbo::invalid_argument_error("Expected array of size ", expected_size,
                                                             ", got array of size ", size);
                    }
                    for (uint32_t i = 0; i < size; ++i) {
                        TURBO_RETURN_NOT_OK(child_converters_[i]->AppendValue(json_obj[i]));
                    }
                    return builder_->append();
                }
                if (json_obj.is_object()) {
                    auto remaining = json_obj.member_count();
                    auto num_children = type_->num_fields();
                    for (int32_t i = 0; i < num_children; ++i) {
                        const auto &field = type_->field(i);
                        auto it = json_obj.find_member(field->name());
                        if (it != json_obj.member_end()) {
                            --remaining;
                            TURBO_RETURN_NOT_OK(child_converters_[i]->AppendValue(it->value));
                        } else {
                            TURBO_RETURN_NOT_OK(child_converters_[i]->append_null());
                        }
                    }
                    if (remaining > 0) {
                        merak::json::StringBuffer sb;
                        merak::json::Writer<merak::json::StringBuffer> writer(sb);
                        json_obj.accept(writer);
                        return turbo::invalid_argument_error("Unexpected members in JSON object for type ",
                                                             type_->to_string(), " Object: ", sb.get_string());
                    }
                    return builder_->append();
                }
                return JSONTypeError("array or object", json_obj.get_type());
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            std::shared_ptr<StructBuilder> builder_;
            std::vector<std::shared_ptr<Converter>> child_converters_;
        };

        // ------------------------------------------------------------------------
        // Converter for union arrays

        class UnionConverter final : public ConcreteConverter<UnionConverter> {
        public:
            explicit UnionConverter(const std::shared_ptr<DataType> &type) { type_ = type; }

            turbo::Status init() override {
                auto union_type = turbo::checked_cast<const UnionType *>(type_.get());
                mode_ = union_type->mode();
                type_id_to_child_num_.clear();
                type_id_to_child_num_.resize(union_type->max_type_code() + 1, -1);
                int child_i = 0;
                for (auto type_id: union_type->type_codes()) {
                    type_id_to_child_num_[type_id] = child_i++;
                }
                std::vector<std::shared_ptr<ArrayBuilder>> child_builders;
                for (const auto &field: type_->fields()) {
                    std::shared_ptr<Converter> child_converter;
                    TURBO_RETURN_NOT_OK(GetConverter(field->type(), &child_converter));
                    child_converters_.push_back(child_converter);
                    child_builders.push_back(child_converter->builder());
                }
                if (mode_ == UnionMode::DENSE) {
                    builder_ = std::make_shared<DenseUnionBuilder>(default_memory_pool(),
                                                                   std::move(child_builders), type_);
                } else {
                    builder_ = std::make_shared<SparseUnionBuilder>(default_memory_pool(),
                                                                    std::move(child_builders), type_);
                }
                return turbo::OkStatus();
            }

            // append a JSON value that must be a 2-long array, containing the type_id
            // and value of the UnionArray's slot.
            turbo::Status AppendValue(const merak::json::Value &json_obj) override {
                if (json_obj.is_null()) {
                    return this->append_null();
                }
                if (!json_obj.is_array()) {
                    return JSONTypeError("array", json_obj.get_type());
                }
                if (json_obj.size() != 2) {
                    return turbo::invalid_argument_error("Expected [type_id, value] pair, got array of size ",
                                                         json_obj.size());
                }
                const auto &id_obj = json_obj[0];
                if (!id_obj.is_int32()) {
                    return JSONTypeError("int", id_obj.get_type());
                }

                auto id = static_cast<int8_t>(id_obj.get_int32());
                auto child_num = type_id_to_child_num_[id];
                if (child_num == -1) {
                    return turbo::invalid_argument_error("type_id ", id, " not found in ", *type_);
                }

                auto child_converter = child_converters_[child_num];
                if (mode_ == UnionMode::SPARSE) {
                    TURBO_RETURN_NOT_OK(turbo::checked_cast<SparseUnionBuilder &>(*builder_).append(id));
                    for (auto &&other_converter: child_converters_) {
                        if (other_converter != child_converter) {
                            TURBO_RETURN_NOT_OK(other_converter->append_null());
                        }
                    }
                } else {
                    TURBO_RETURN_NOT_OK(turbo::checked_cast<DenseUnionBuilder &>(*builder_).append(id));
                }
                return child_converter->AppendValue(json_obj[1]);
            }

            std::shared_ptr<ArrayBuilder> builder() override { return builder_; }

        private:
            UnionMode::type mode_;
            std::shared_ptr<ArrayBuilder> builder_;
            std::vector<std::shared_ptr<Converter>> child_converters_;
            std::vector<int8_t> type_id_to_child_num_;
        };

        // ------------------------------------------------------------------------
        // General conversion functions

        turbo::Status ConversionNotImplemented(const std::shared_ptr<DataType> &type) {
            return turbo::unimplemented_error("JSON conversion to ", type->to_string(),
                                              " not implemented");
        }

        turbo::Status GetDictConverter(const std::shared_ptr<DataType> &type,
                                       std::shared_ptr<Converter> *out) {
            std::shared_ptr<Converter> res;

            const auto value_type = turbo::checked_cast<const DictionaryType &>(*type).get_value_type();

            switch (value_type->id()) {
                case Type::INT8:
                    res = std::make_shared<IntegerConverter<Int8Type, DictionaryBuilder<Int8Type>>>(type);
                    break;
                case Type::INT16:
                    res = std::make_shared<IntegerConverter<Int16Type, DictionaryBuilder<Int16Type>>>(type);
                    break;
                case Type::INT32:
                    res = std::make_shared<IntegerConverter<Int32Type, DictionaryBuilder<Int32Type>>>(type);
                    break;
                case Type::INT64:
                    res = std::make_shared<IntegerConverter<Int64Type, DictionaryBuilder<Int64Type>>>(type);
                    break;
                case Type::UINT8:
                    res = std::make_shared<IntegerConverter<UInt8Type, DictionaryBuilder<UInt8Type>>>(type);
                    break;
                case Type::UINT16:
                    res = std::make_shared<IntegerConverter<UInt16Type, DictionaryBuilder<UInt16Type>>>(type);
                    break;
                case Type::UINT32:
                    res = std::make_shared<IntegerConverter<UInt32Type, DictionaryBuilder<UInt32Type>>>(type);
                    break;
                case Type::UINT64:
                    res = std::make_shared<IntegerConverter<UInt64Type, DictionaryBuilder<UInt64Type>>>(type);
                    break;
                case Type::FP32:
                    res = std::make_shared<FloatConverter<Fp32Type, DictionaryBuilder<Fp32Type>>>(type);
                    break;
                case Type::FP64:
                    res = std::make_shared<FloatConverter<Fp64Type, DictionaryBuilder<Fp64Type>>>(type);
                    break;
                case Type::STRING:
                    res = std::make_shared<StringConverter<StringType, DictionaryBuilder<StringType>>>(type);
                    break;
                case Type::BINARY:
                    res = std::make_shared<StringConverter<BinaryType, DictionaryBuilder<BinaryType>>>(type);
                    break;
                case Type::LARGE_STRING:
                    res = std::make_shared<StringConverter<LargeStringType, DictionaryBuilder<LargeStringType>>>(type);
                    break;
                case Type::LARGE_BINARY:
                    res = std::make_shared<StringConverter<LargeBinaryType, DictionaryBuilder<LargeBinaryType>>>(type);
                    break;
                case Type::STRING_VIEW:
                    res = std::make_shared<StringConverter<StringViewType, DictionaryBuilder<StringViewType>>>(type);
                    break;
                case Type::BINARY_VIEW:
                    res = std::make_shared<StringConverter<BinaryViewType, DictionaryBuilder<BinaryViewType>>>(type);
                    break;
                case Type::FIXED_SIZE_BINARY:
                    res = std::make_shared<FixedSizeBinaryConverter<DictionaryBuilder<FixedSizeBinaryType>>>(type);
                    break;
                case Type::DECIMAL128:
                    res = std::make_shared<Decimal128Converter<DictionaryBuilder<Decimal128Type>>>(type);
                    break;
                case Type::DECIMAL256:
                    res = std::make_shared<Decimal256Converter<DictionaryBuilder<Decimal256Type>>>(type);
                    break;
                default:
                    return ConversionNotImplemented(type);
            }
            TURBO_RETURN_NOT_OK(res->init());
            *out = res;
            return turbo::OkStatus();
        }

        turbo::Status GetConverter(const std::shared_ptr<DataType> &type,
                                   std::shared_ptr<Converter> *out) {
            if (type->id() == Type::DICTIONARY) {
                return GetDictConverter(type, out);
            }

            std::shared_ptr<Converter> res;

            switch (type->id()) {
                case Type::INT8:
                    res = std::make_shared<IntegerConverter<Int8Type>>
                            (type);
                    break;
                case Type::INT16:
                    res = std::make_shared<IntegerConverter<Int16Type>>
                            (type);
                    break;
                case Type::INT32:
                    res = std::make_shared<IntegerConverter<Int32Type>>
                            (type);
                    break;
                case Type::INT64:
                    res = std::make_shared<IntegerConverter<Int64Type>>
                            (type);
                    break;
                case Type::UINT8:
                    res = std::make_shared<IntegerConverter<UInt8Type>>
                            (type);
                    break;
                case Type::UINT16:
                    res = std::make_shared<IntegerConverter<UInt16Type>>
                            (type);
                    break;
                case Type::UINT32:
                    res = std::make_shared<IntegerConverter<UInt32Type>>
                            (type);
                    break;
                case Type::UINT64:
                    res = std::make_shared<IntegerConverter<UInt64Type>>
                            (type);
                    break;
                case Type::TIMESTAMP:
                    res = std::make_shared<TimestampConverter>(type);
                    break;
                case Type::DATE32:
                    res = std::make_shared<IntegerConverter<Date32Type>>
                            (type);
                    break;
                case Type::DATE64:
                    res = std::make_shared<IntegerConverter<Date64Type>>
                            (type);
                    break;
                case Type::TIME32:
                    res = std::make_shared<IntegerConverter<Time32Type>>
                            (type);
                    break;
                case Type::TIME64:
                    res = std::make_shared<IntegerConverter<Time64Type>>
                            (type);
                    break;
                case Type::DURATION:
                    res = std::make_shared<IntegerConverter<DurationType>>
                            (type);
                    break;
                case Type::NA:
                    res = std::make_shared<NullConverter>(type);
                    break;
                case Type::BOOL:
                    res = std::make_shared<BooleanConverter>(type);
                    break;
                case Type::FP16:
                    res = std::make_shared<IntegerConverter<Fp16Type>>
                            (type);
                    break;
                case Type::FP32:
                    res = std::make_shared<FloatConverter<Fp32Type>>
                            (type);
                    break;
                case Type::FP64:
                    res = std::make_shared<FloatConverter<Fp64Type>>
                            (type);
                    break;
                case Type::LIST:
                    res = std::make_shared<VarLengthListLikeConverter<ListType>>
                            (type);
                    break;
                case Type::LARGE_LIST:
                    res = std::make_shared<VarLengthListLikeConverter<LargeListType>>
                            (type);
                    break;
                case Type::LIST_VIEW:
                    res = std::make_shared<VarLengthListLikeConverter<ListViewType>>
                            (type);
                    break;
                case Type::LARGE_LIST_VIEW:
                    res = std::make_shared<VarLengthListLikeConverter<LargeListViewType>>
                            (type);
                    break;
                case Type::MAP:
                    res = std::make_shared<MapConverter>(type);
                    break;
                case Type::FIXED_SIZE_LIST:
                    res = std::make_shared<FixedSizeListConverter>(type);
                    break;
                case Type::STRUCT:
                    res = std::make_shared<StructConverter>(type);
                    break;
                case Type::STRING:
                    res = std::make_shared<StringConverter<StringType>>
                            (type);
                    break;
                case Type::BINARY:
                    res = std::make_shared<StringConverter<BinaryType>>
                            (type);
                    break;
                case Type::LARGE_STRING:
                    res = std::make_shared<StringConverter<LargeStringType>>
                            (type);
                    break;
                case Type::LARGE_BINARY:
                    res = std::make_shared<StringConverter<LargeBinaryType>>
                            (type);
                    break;
                case Type::STRING_VIEW:
                    res = std::make_shared<StringConverter<StringViewType>>
                            (type);
                    break;
                case Type::BINARY_VIEW:
                    res = std::make_shared<StringConverter<BinaryViewType>>
                            (type);
                    break;
                case Type::FIXED_SIZE_BINARY:
                    res = std::make_shared<FixedSizeBinaryConverter<>>(type);
                    break;
                case Type::DECIMAL128:
                    res = std::make_shared<Decimal128Converter<>>(type);
                    break;
                case Type::DECIMAL256:
                    res = std::make_shared<Decimal256Converter<>>(type);
                    break;
                case Type::SPARSE_UNION:
                    res = std::make_shared<UnionConverter>(type);
                    break;
                case Type::DENSE_UNION:
                    res = std::make_shared<UnionConverter>(type);
                    break;
                case Type::INTERVAL_MONTHS:
                    res = std::make_shared<IntegerConverter<MonthIntervalType>>
                            (type);
                    break;
                case Type::INTERVAL_DAY_TIME:
                    res = std::make_shared<DayTimeIntervalConverter>(type);
                    break;
                case Type::INTERVAL_MONTH_DAY_NANO:
                    res = std::make_shared<MonthDayNanoIntervalConverter>(type);
                    break;
                default:
                    return ConversionNotImplemented(type);
            }

            TURBO_RETURN_NOT_OK(res->init());
            *out = res;
            return turbo::OkStatus();
        }

    }  // namespace

    turbo::Result<std::shared_ptr<Array>> array_from_json(const std::shared_ptr<DataType> &type,
                                                        std::string_view json_string) {
        std::shared_ptr<Converter> converter;
        TURBO_RETURN_NOT_OK(GetConverter(type, &converter));

        merak::json::Document json_doc;
        json_doc.parse<kParseFlags>(json_string.data(), json_string.length());
        if (json_doc.has_parse_error()) {
            return turbo::invalid_argument_error("JSON parse error at offset ", json_doc.get_error_offset(), ": ",
                                                 get_parse_error_en(json_doc.get_parse_error()));
        }

        // The JSON document should be an array, append it
        TURBO_RETURN_NOT_OK(converter->append_values(json_doc));
        std::shared_ptr<Array> out;
        TURBO_RETURN_NOT_OK(converter->finish(&out));
        return out;
    }

    turbo::Status chunked_array_from_json(const std::shared_ptr<DataType> &type,
                                       const std::vector<std::string> &json_strings,
                                       std::shared_ptr<ChunkedArray> *out) {
        ArrayVector out_chunks;
        out_chunks.reserve(json_strings.size());
        for (const std::string &chunk_json: json_strings) {
            out_chunks.emplace_back();
            TURBO_MOVE_OR_RAISE(out_chunks.back(), array_from_json(type, chunk_json));
        }
        *out = std::make_shared<ChunkedArray>(std::move(out_chunks), type);
        return turbo::OkStatus();
    }

    turbo::Status dict_array_from_json(const std::shared_ptr<DataType> &type,
                                    std::string_view indices_json, std::string_view dictionary_json,
                                    std::shared_ptr<Array> *out) {
        if (type->id() != Type::DICTIONARY) {
            return turbo::failed_precondition_error("dict_array_from_json requires dictionary type, got ", *type);
        }

        const auto &dictionary_type = turbo::checked_cast<const DictionaryType &>(*type);

        TURBO_MOVE_OR_RAISE(auto indices,
                            array_from_json(dictionary_type.index_type(), indices_json));
        TURBO_MOVE_OR_RAISE(auto dictionary,
                            array_from_json(dictionary_type.get_value_type(), dictionary_json));

        return DictionaryArray::from_arrays(type, std::move(indices), std::move(dictionary))
                .try_value(out);
    }

    turbo::Result<std::shared_ptr<Array>> dict_array_from_json(const std::shared_ptr<DataType> &type, std::string_view indices_json,
                                   std::string_view dictionary_json) {
        std::shared_ptr<Array> r;
        TURBO_RETURN_NOT_OK(dict_array_from_json(type, indices_json, dictionary_json, &r));
        return r;
    }

    turbo::Status scalar_from_json(const std::shared_ptr<DataType> &type, std::string_view json_string,
                                 std::shared_ptr<Scalar> *out) {
        std::shared_ptr<Converter> converter;
        TURBO_RETURN_NOT_OK(GetConverter(type, &converter));

        merak::json::Document json_doc;
        json_doc.parse<kParseFlags>(json_string.data(), json_string.length());
        if (json_doc.has_parse_error()) {
            return turbo::invalid_argument_error("JSON parse error at offset ", json_doc.get_error_offset(), ": ",
                                                 get_parse_error_en(json_doc.get_parse_error()));
        }

        std::shared_ptr<Array> array;
        TURBO_RETURN_NOT_OK(converter->AppendValue(json_doc));
        TURBO_RETURN_NOT_OK(converter->finish(&array));
        DKCHECK_EQ(array->length(), 1);
        return array->get_scalar(0).try_value(out);
    }

    turbo::Status dict_scalar_from_json(const std::shared_ptr<DataType> &type,
                                     std::string_view index_json, std::string_view dictionary_json,
                                     std::shared_ptr<Scalar> *out) {
        if (type->id() != Type::DICTIONARY) {
            return turbo::failed_precondition_error("dict_scalar_from_json requires dictionary type, got ", *type);
        }

        const auto &dictionary_type = turbo::checked_cast<const DictionaryType &>(*type);

        std::shared_ptr<Scalar> index;
        std::shared_ptr<Array> dictionary;
        TURBO_RETURN_NOT_OK(scalar_from_json(dictionary_type.index_type(), index_json, &index));
        TURBO_MOVE_OR_RAISE(dictionary,
                            array_from_json(dictionary_type.get_value_type(), dictionary_json));

        *out = DictionaryScalar::create(std::move(index), std::move(dictionary));
        return turbo::OkStatus();
    }
}  // namespace nebula::ipc::json
